#include "pub_sub_common.h"

ipc::route* p_shm_channel = nullptr;

// Signal handler function
void signal_handler(int signal) {
  if (p_shm_channel != nullptr) {
    p_shm_channel->disconnect();
  }
}

// Function to run the nanomsg publisher
void nanomsg_publisher() {
  ipc::route shm_channel{kIpcRouteName, ipc::sender};
  if (!shm_channel.valid()) {
    fprintf(stderr, "channel %s is not valid.\n", kIpcRouteName);
    return;
  } else {
    p_shm_channel = &shm_channel;
  }
  shm_channel.wait_for_recv(1);
  fprintf(stdout, "channel recv_count is %d.\n", static_cast<uint32_t>(shm_channel.recv_count()));
  setup_signal_handlers();

  int sock = nn_socket(AF_SP, NN_PUB);
  if (sock < 0) {
    fprintf(stderr, "Error creating publisher socket: %s\n", nn_strerror(nn_errno()));
    return;
  }
  if (nn_bind(sock, URL) < 0) {
    fprintf(stderr, "Error binding publisher socket: %s\n", nn_strerror(nn_errno()));
    nn_close(sock);
    return;
  }
  int read_buf_len{0};
  size_t sz = sizeof (read_buf_len);
  nn_getsockopt(sock, NN_SOL_SOCKET, NN_SNDBUF, &read_buf_len, &sz);
  fprintf(stdout, "### Default NN_SNDBUF size %d.\n", read_buf_len);
  fprintf(stdout, "Publisher socket bound to %s\n", URL);
  std::this_thread::sleep_for(std::chrono::seconds(1));

  // Size of each shm message
  constexpr int SHM_MESSAGE_SIZE = 32;
  char message[SHM_MESSAGE_SIZE] = "nanomsg ipc message";

  std::vector<uint8_t> large_data;
  uint32_t data_size = create_large_data(large_data);
  uint16_t crc = calculate_crc({large_data.begin() + sizeof(uint32_t) + sizeof(uint16_t), large_data.end()});

  std::memcpy(large_data.data(), &data_size, sizeof(data_size));
  std::memcpy(large_data.data() + sizeof(data_size), &crc, sizeof(crc));

  fprintf(stdout, "@@ Start to send, current time: %s\n", get_currenttime_millis().c_str());
  for (int i = 0; i < NUM_MESSAGES; ++i) {
    if (shm_channel.send(large_data.data(), large_data.size())) {
  #if DEBUG_ON
      fprintf(stdout, "Producer sent large data of size %lu bytes (data_size: %u, crc: %hx)\n", large_data.size(),
              data_size, crc);
  #endif
    } else {
      fprintf(stdout, "Producer sent failed of size %lu bytes (data_size: %u, crc: %hx)\n", large_data.size(),
              data_size, crc);
    }

    int bytes = nn_send(sock, message, SHM_MESSAGE_SIZE, 0);
    if (bytes < 0) {
      fprintf(stderr, "Error sending message: %s\n", nn_strerror(nn_errno()));
      return;
    }
    std::this_thread::sleep_for(std::chrono::microseconds(THROTTLE_MESSAGE_SLEEP));  // Throttle messages
  }
  fprintf(stdout, "@@ Finish send, current time: %s\n", get_currenttime_millis().c_str());
  std::this_thread::sleep_for(std::chrono::seconds(3));
  nn_close(sock);
}

int main(int argc, char* argv[]) {
  nanomsg_publisher();
  fprintf(stdout, "Publisher finish send %d messages\n", NUM_MESSAGES);
  return 0;
}
